package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo3MysqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source 读取mysql中的数据
    val mysqlDS: DataStream[String] = env.addSource(new MysqlSource)
    mysqlDS.print()

    env.execute()

  }

}

/**
 * 自定义source 是 flink对外提供的一个接口，理论上可以读取任何数据源的数据
 *
 */
class MysqlSource extends SourceFunction[String] {
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //2、创建链接
    val con: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/student", "root", "")
    //3、写sql查询数据
    val stat: PreparedStatement = con.prepareStatement("select * from student")
    //4、执行查询
    val resultSet: ResultSet = stat.executeQuery()
    //5、解析数据
    while (resultSet.next()) {
      //通过字段的类型和字段的名称解析数据
      val id: Long = resultSet.getLong("id")
      val name: String = resultSet.getString("name")
      val age: Long = resultSet.getLong("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")

      //将数据发送到下游
      ctx.collect(s"$id\t$name\t$age\t$gender\t$clazz")

    }

    //关闭链接
    stat.close()
    con.close()
  }

  override def cancel(): Unit = {}
}
